"""
Case Type   : Logical_Replication
Case Name   : 开启极致RTO时，备机逻辑解码失败，合理报错
Create At   : 2024/1/15
Owner       : xiaqi14
Description :
    1.创建目录，修改参数
    2.配置初始用户白名单
    3.开启极致RTO
    4.创建表
    5.在主机上创建逻辑复制槽
    6.主机和备机分别开启逻辑复制槽
    7.执行DML语句解码
    8.停止逻辑复制槽
    9.删除逻辑复制槽
    10.资源清理
    11.恢复参数，恢复pg_hba.conf文件
Expect      :
    1.创建目录成功，修改参数成功
    2.配置成功
    3.开启极致RTO成功
    4.创建成功
    5.创建成功
    6.主机开启成功，备机开启失败
    7.执行DML语句成功，主机解码成功
    8.停止成功
    9.删除成功
    10.清理成功
    11.恢复成功
History     :
"""
import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread

primary_sh = CommonSH('PrimaryDbUser')


@unittest.skipIf(1 == primary_sh.get_node_num(), '单机环境不执行')
class LogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.standby_node = Node('Standby1DbUser')
        self.sh_standby = CommonSH('Standby1DbUser')
        self.constant = Constant()
        self.common = Common()
        self.dir_replication = os.path.join(macro.DB_BACKUP_PATH,
                                            'dir_logical_replication_0075/')
        self.table_name = 't_logical_replication_0075'
        self.slot_name_1 = 'slot_logical_replication_0075_1'
        self.slot_name_2 = 'slot_logical_replication_0075_2'
        self.logfile = os.path.join(self.dir_replication,
                                    'logfile_logical_replication_0075.log')
        self.config_slot_log = 'enable_slot_log'
        self.slot_log_default = self.common.show_param(self.config_slot_log)
        self.config_wal_level = 'wal_level'
        self.wal_level_default = self.common.show_param(self.config_wal_level)
        self.replication_type = 'replication_type'
        self.replication_type_default = \
            self.common.show_param(self.replication_type)
        self.recovery_parse_workers = 'recovery_parse_workers'
        self.parse_workers_default = \
            self.common.show_param(self.recovery_parse_workers)
        self.recovery_redo_workers = 'recovery_redo_workers'
        self.redo_workers_default = \
            self.common.show_param(self.recovery_redo_workers)
        self.hot_standby = 'hot_standby'
        self.hot_standby_default = self.common.show_param(self.hot_standby)

    def test_logical_replication(self):
        text = "-----step1:创建目录，修改参数;expect:创建目录成功，" \
               "修改参数成功-----"
        self.logger.info(text)
        mkdir_cmd = f'''rm -rf {self.dir_replication} &&
                        mkdir {self.dir_replication} &&
                        ls -ld {self.dir_replication}'''
        primary_msg = self.primary_node.sh(mkdir_cmd).result()
        self.logger.info(primary_msg)
        self.assertIn(self.dir_replication, primary_msg, '执行失败' + text)

        if 'on' != self.slot_log_default:
            guc_result = primary_sh.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.common.show_param(self.config_slot_log)
        self.assertEquals(show_msg, 'on', '执行失败' + text)

        if 'logical' != self.wal_level_default:
            guc_result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)

        text = "-----step2:配置初始用户白名单;expect:配置成功-----"
        self.logger.info(text)
        param_1 = f'local replication {self.primary_node.ssh_user}  trust'
        param_2 = f'host  replication {self.primary_node.ssh_user} ' \
            f'127.0.0.1/32 trust'
        param_3 = f'host  replication {self.primary_node.ssh_user} ' \
            f'::1/128 trust'
        result_1 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        text = "-----step3:开启极致RTO，expect：开启极致RTO成功-----"
        self.logger.info(text)
        if '1' != self.replication_type_default:
            guc_result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.replication_type}=1')
            self.assertTrue(guc_result, '执行失败' + text)
        if '4' != self.parse_workers_default:
            guc_result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.recovery_parse_workers}=4')
            self.assertTrue(guc_result, '执行失败' + text)
        if '4' != self.redo_workers_default:
            guc_result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.recovery_redo_workers}=4')
            self.assertTrue(guc_result, '执行失败' + text)
        if 'off' != self.hot_standby_default:
            guc_result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.hot_standby}=off')
            self.assertTrue(guc_result, '执行失败' + text)
        restart_msg = primary_sh.restart_db_cluster()
        self.logger.info(restart_msg)
        self.assertTrue(restart_msg, '重启失败')
        status = primary_sh.get_db_cluster_status('detail')
        self.assertTrue("Degraded" in status or "Normal" in status,
                        '执行失败:' + text)
        show_msg = self.common.show_param(self.config_wal_level)
        self.assertEquals(show_msg, 'logical', '执行失败' + text)
        show_msg = self.common.show_param(self.replication_type)
        self.assertEquals(show_msg, '1', '执行失败' + text)
        show_msg = self.common.show_param(self.recovery_parse_workers)
        self.assertEquals(show_msg, '4', '执行失败' + text)
        show_msg = self.common.show_param(self.recovery_redo_workers)
        self.assertEquals(show_msg, '4', '执行失败' + text)
        show_msg = self.common.show_param(self.hot_standby)
        self.assertEquals(show_msg, 'off', '执行失败' + text)

        text = "-----step4:创建表;expect:创建成功-----"
        self.logger.info(text)
        create_cmd = f'''drop table if exists {self.table_name};
            create table {self.table_name}(
                col_integer integer,
                col_date date,
                col_varchar varchar(20));'''
        create_msg = primary_sh.execut_db_sql(create_cmd, dbname='postgres')
        self.logger.info(create_msg)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS and
                      self.constant.CREATE_TABLE_SUCCESS,
                      create_msg, '执行失败:' + text)

        text = "-----step5:在主机上创建逻辑复制槽;expect:创建成功-----"
        self.logger.info(text)
        slot_names = [self.slot_name_1, self.slot_name_2]
        for slot_name in slot_names:
            create_slot_msg = primary_sh.exec_pg_recvlogical(
                '--create', 'postgres', slot_name,
                self.primary_node.ssh_user, '-s 2 ')
            self.logger.info(create_slot_msg)
        text = "-----验证创建逻辑复制槽成功;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"select * from pg_get_replication_slots() " \
            f"where slot_name like '%slot_logical_replication_0075%';"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1 and self.slot_name_2,
                      sql_msg, '执行失败:' + text)

        text = "-----step6:主机和备机分别开启逻辑复制槽;" \
               "expect:主机开启成功，备机开启失败-----"
        self.logger.info(text)
        thread_1 = ComThread(primary_sh.exec_pg_recvlogical,
                             args=('--start', 'postgres', self.slot_name_1,
                                   self.primary_node.ssh_user,
                                   f'-f {self.logfile} '
                                   f'-o skip-empty-xacts=true -F 1 -s 2 ',))
        thread_1.setDaemon(True)
        thread_1.start()
        thread_1.join(10)
        self.logger.info(thread_1.get_result())
        text = "-----验证主机开启逻辑复制槽成功;expect:开启成功-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_1}%';;"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertEquals('t', sql_msg.splitlines()[-2].strip(),
                          '执行失败:' + text)

        thread_2 = ComThread(self.sh_standby.exec_pg_recvlogical,
                             args=('--start', 'postgres', self.slot_name_2,
                                   self.primary_node.ssh_user,
                                   f'-f - -o skip-empty-xacts=true -F 1 -s 2',))
        thread_2.setDaemon(True)
        thread_2.start()
        thread_2.join(10)
        self.logger.info(thread_2.get_result())
        text = "-----验证备机开启逻辑复制槽失败;expect:开启失败-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_2}%';;"
        sql_msg = self.sh_standby.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn("FATAL:  can not accept connection if hot standby off",
                      sql_msg, '执行失败:' + text)

        text = "-----step7:执行DML语句解码;expect:执行DML语句成功----"
        self.logger.info(text)
        sql_cmd = f'''insert into {self.table_name} 
            values(1,date '2022-05-22','测试');
            update {self.table_name} set col_integer=3,col_varchar='开发';
            delete from {self.table_name};'''
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG and
                      self.constant.UPDATE_SUCCESS_MSG and
                      self.constant.DELETE_SUCCESS_MSG,
                      sql_msg, "执行失败" + text)

    def tearDown(self):
        text = "-----step8:停止逻辑复制槽;expect:停止成功-----"
        self.logger.info(text)
        stop_cmd = "ps -ef | grep  pg_recvlogical | grep -v grep | " \
                   "awk '{{print $2}}' | xargs kill -9"
        self.logger.info(stop_cmd)
        primary_result = self.primary_node.sh(stop_cmd).result()
        self.logger.info(primary_result)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '%{self.slot_name_1}%';"
        primary_active_msg = primary_sh.execut_db_sql(sql_cmd,
                                                      dbname='postgres')
        self.logger.info(primary_active_msg)

        text_1 = "-----step9:删除逻辑复制槽;expect:删除成功-----"
        self.logger.info(text_1)
        slot_names = [self.slot_name_1, self.slot_name_2]
        for slot_name in slot_names:
            drop_slot_msg = primary_sh.exec_pg_recvlogical(
                '--drop', 'postgres', slot_name,
                self.primary_node.ssh_user, '-s 2 ')
            self.logger.info(drop_slot_msg)
        sql_cmd = f"select * from pg_get_replication_slots() " \
            f"where slot_name like '%slot_logical_replication_0075%';"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)

        text_2 = "-----step10:资源清理;expect:清理成功-----"
        self.logger.info(text_2)
        drop_cmd = f'''drop table {self.table_name} cascade;'''
        drop_msg = primary_sh.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)
        del_cmd = f'rm -rf {self.dir_replication};'
        self.logger.info(del_cmd)
        primary_del_res = self.primary_node.sh(del_cmd).result()
        self.logger.info(primary_del_res)
        file_cmd = f'''if [ -d {self.dir_replication} ];
            then echo "exists"; else echo "not exists"; fi'''
        primary_file_msg = self.primary_node.sh(file_cmd).result()

        text_3 = "-----step11:恢复参数，恢复pg_hba.conf文件，expect:恢复成功-----"
        self.logger.info(text_3)
        slot_log_current = self.common.show_param(self.config_slot_log)
        if slot_log_current != self.slot_log_default:
            primary_sh.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_slot_log}={self.slot_log_default}')
        show_slot_log = self.common.show_param(self.config_slot_log)

        wal_level_current = self.common.show_param(self.config_wal_level)
        if wal_level_current != self.wal_level_default:
            primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}={self.wal_level_default}')

        param_1 = f'local replication {self.primary_node.ssh_user} '
        param_2 = f'host replication {self.primary_node.ssh_user} 127.0.0.1/32 '
        param_3 = f'host  replication {self.primary_node.ssh_user} ::1/128 '
        result_1 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)

        replication_type_current = self.common.show_param(self.replication_type)
        if '1' != replication_type_current:
            guc_result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.replication_type}=1')
            self.logger.info(guc_result)

        parse_workers_current = \
            self.common.show_param(self.recovery_parse_workers)
        if '1' != parse_workers_current:
            guc_result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.recovery_parse_workers}=1')
            self.logger.info(guc_result)

        redo_workers_current = \
            self.common.show_param(self.recovery_redo_workers)
        if '1' != redo_workers_current:
            guc_result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.recovery_redo_workers}=1')
            self.logger.info(guc_result)

        hot_standby_current = self.common.show_param(self.hot_standby)
        if 'on' != hot_standby_current:
            guc_result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.hot_standby}=on')
            self.logger.info(guc_result)

        restart_msg = primary_sh.restart_db_cluster()
        self.logger.info(restart_msg)
        status = primary_sh.get_db_cluster_status('detail')
        self.logger.info(status)
        show_wal_level = self.common.show_param(self.config_wal_level)
        show_replication_type = self.common.show_param(self.replication_type)
        show_parse_workers = self.common.show_param(self.recovery_parse_workers)
        show_redo_workers = self.common.show_param(self.recovery_redo_workers)
        show_hot_standby = self.common.show_param(self.hot_standby)

        self.assertEquals('f', primary_active_msg.splitlines()[-2].strip(),
                          '执行失败:' + text_3)
        self.assertIn("(0 rows)", sql_msg, '执行失败:' + text_1)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS,
                      drop_msg, '执行失败' + text_2)
        self.assertEqual("not exists", primary_file_msg, '执行失败' + text_2)
        self.assertEquals(show_slot_log, self.slot_log_default,
                          '执行失败' + text_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text_3)
        self.assertTrue(restart_msg, '重启失败')
        self.assertTrue("Degraded" in status or "Normal" in status,
                        '执行失败:' + text_3)
        self.assertEquals(show_wal_level, self.wal_level_default,
                          '执行失败' + text)
        self.assertEquals(show_replication_type, '1', '执行失败' + text)
        self.assertEquals(show_parse_workers, '1', '执行失败' + text)
        self.assertEquals(show_redo_workers, '1', '执行失败' + text)
        self.assertEquals(show_hot_standby, 'on', '执行失败' + text)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
